package com.ibs.hadoop.spring.hbase;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Service;

import com.ibs.hadoop.spring.hbase.bean.PageBean;

/**
 * hbase服务操作
* 源文件名：HbaseParent.java
* 文件版本：1.0.0
* 创建作者：liujun
* 创建日期：2016年9月7日
* 修改作者：liujun
* 修改日期：2016年9月7日
* 文件描述：TODO
* 版权所有：Copyright 2016 zjhz, Inc. All Rights Reserved.
*/
@Service
public class HbaseParent {

    /**
     * hbase操作对象
     * @author liujun
    * @字段说明 htemplate
    */
    @Autowired
    private HbaseTemplate htemplate;

    /**
     * 通过rowkey，获取指定列族下的所有数据
    * 方法描述
    * @param table 表名
    * @param family 列族
    * @param qualifier 限定名
    * @param rowKey 主键
    * @param spitFlag 分隔符
    * @return
    * @创建日期 2016年9月7日
    */
    public List<String[]> getAllValueByRowKey(String table, String family, String qualifier, String rowKey,
            String spitFlag) throws Exception {

        final List<String[]> hbaseResult = new ArrayList<String[]>();

        // 通过rowkey获取数据
        this.htemplate.get(table, rowKey, family, qualifier, (Result result, int rowNum) -> {
            if (null != result && result.listCells() != null) {
                // 将结果放入到map中
                result.listCells().forEach(cell -> {
                    String[] valueArray = new String(CellUtil.cloneValue(cell)).split(spitFlag);
                    hbaseResult.add(valueArray);
                });
            }

            return hbaseResult;
        });

        return hbaseResult;
    }

    /**
     * 通过行键进行数据的过滤操作
    * 方法描述
    * @param table 表名
    * @param family 列族
    * @param qualifier 列限定名 
    * @param rowKey 行键
    * @param lineSpitFlag 行数据分隔符
    * @param page 分页信息
    * @return
    * @创建日期 2016年9月7日
    */
    public List<String[]> findValueByRowKey(String table, String rowKey, String lineSpitFlag, PageBean page) {

        final List<String[]> hbaseResult = new ArrayList<String[]>();

        Scan scan = new Scan();

        // 通过过滤器过滤数据
        scan.setStartRow(page.getUpOverKey().getBytes());
        scan.setFilter(new PageFilter(page.getPageSize()));
        scan.setMaxResultSize(page.getPageSize());

        this.htemplate.find(table, scan, (ResultScanner results) -> {
            results.forEach(s -> {
                if (null != s && s.listCells() != null) {
                    List<Cell> cellLists = s.listCells();

                    cellLists.forEach(cell -> {
                        String[] valueArray = new String(CellUtil.cloneValue(cell)).split(lineSpitFlag);
                        hbaseResult.add(valueArray);
                    });
                }
            });

            return hbaseResult;
        });

        return hbaseResult;
    }
}
